package cn.itcast.streaming.function.window;


import cn.itcast.streaming.bean.ElectricFenceModel;
import com.google.common.collect.Lists;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedList;

//对电子围栏进行自定义窗口操作，处理电子围栏判断逻辑（对电子围栏模型数据进行再次拉宽）
// 状态报警(0:出围栏  1:进围栏） //进围栏时间   //出围栏时间

public class ElectricFenceWindowFunction extends RichWindowFunction<ElectricFenceModel,ElectricFenceModel,String, TimeWindow> {
    //定义存储历史电子围栏状态的state对象
    private MapState<String,Integer> state = null;
    private String stateStartWith = "electricFence_";
    //
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //定义mapState的描述器（相当于定义表结构）
        MapStateDescriptor<String, Integer> mapStateDescriptor = new MapStateDescriptor<>(
                "electricFenceState",
                TypeInformation.of(new TypeHint<String>() {}),
                TypeInformation.of(new TypeHint<Integer>() {})
        );
        ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        //对state数据设置有效期
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.seconds(Long.parseLong(
                parameterTool.getRequired("ruleinfo.millionseconds"))))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)//每次数据更新的时候同时更新ttl time
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)//永不返回过期数据
                .build();

        //设置state过期策略
        mapStateDescriptor.enableTimeToLive(ttlConfig);

        //实例化mapState对象
        state = getRuntimeContext().getMapState(mapStateDescriptor);
    }

    @Override
    public void apply(String key, TimeWindow timeWindow, Iterable<ElectricFenceModel> iterable, Collector<ElectricFenceModel> collector) throws Exception {

        //排序
        LinkedList<ElectricFenceModel> electricFenceModels = Lists.newLinkedList(iterable);
        Collections.sort(electricFenceModels);


        //TODO 2.从state中获取车辆vin对应的flag标记(车辆在围栏中或不在) 0：电子围栏里面 1：电子围栏外面
        //需要将当前的行驶的位置（围栏内还是围栏外）作为下一条数据的一个属性，所以将当前的状态保存到state中，
        // 可以利用flink——ValueState将上一次的历史状态存储到ValueState中
        Integer lastStateValue = state.get(stateStartWith + key);
        System.out.println("上一个窗口的电子围栏状态数据："+lastStateValue);
        if(lastStateValue==null){
            lastStateValue = -999999;
        }

        //判断当前处于内还是外
        long electricFenceInCount = electricFenceModels.stream().filter(efMode -> (efMode.getNowStatus() == 0)).count();
        long electricFenceOutCount = electricFenceModels.stream().filter(efMode -> (efMode.getNowStatus() == 1)).count();

        //定义电子围栏状态（0：内，1：外）
        int currentStateValue = 1;
        if(electricFenceInCount>=electricFenceOutCount) currentStateValue = 0;

        //将当前状态数据写入state
        state.put(stateStartWith+key,currentStateValue);

        //判断历史状态数据与当前状态是否有区别
        if(lastStateValue != currentStateValue){
            //说明状态发生改变
            if((lastStateValue ==1 || lastStateValue == -999999) && currentStateValue ==0){ //进入围栏判断

                //进入电子围栏(获取第一条进入围栏的数据
                ElectricFenceModel firstInFenceModel = electricFenceModels.stream().filter(
                        efMode -> efMode.getNowStatus() == 0).findFirst().get();
                firstInFenceModel.setInEleTime(firstInFenceModel.getTerminalTime());
                //状态报警(0:出围栏  1:进围栏）
                firstInFenceModel.setStatusAlarm(1);
                collector.collect(firstInFenceModel);
            }else if(lastStateValue == 0 && currentStateValue ==1){ //出围栏判断
                //取最后一条在围栏外的数据最后
                ElectricFenceModel finalOutFenceModel  = electricFenceModels.stream().filter(
                        efMode -> efMode.getNowStatus() == 1).sorted(Comparator.reverseOrder()).findFirst().get();
                finalOutFenceModel.setOutEleTime(finalOutFenceModel.getTerminalTime());
                finalOutFenceModel.setStatusAlarm(0);
                collector.collect(finalOutFenceModel);
            }
        }
    }
}
